package cn.ipanel.bigdata.dw.dim.phoenix

import cn.ipanel.bigdata.boot.date.Time
import cn.ipanel.bigdata.boot.source.genre.Phoenix
import cn.ipanel.bigdata.dw.BaseColumns
import cn.ipanel.bigdata.dw.dws.HBASE_NAMESPACE
import org.apache.spark.sql.{DataFrame, Dataset, Row}

/**
 * @author lzz
 * @environment IntelliJ IDEA 2020.3.1
 * @projectName bigdata_shanxi
 * @date 2023/08/04 15:00
 * @description: 任务运行统计，包括实时任务
 */
 protected[phoenix] object TaskStat extends Phoenix(HBASE_NAMESPACE, "t_task_stat") with BaseColumns {

  final val F_TASK_NAME     : String = "f_task_name"              // 任务名称
  final val F_TABLE_NAME    : String = "f_table_name"             // 表名称，对于不需要区分表名的，默认给 default。可以手动给表名字，像实时任务可能有多个表的，但需要显示调用
  final val F_APP_ID        : String = "f_app_id"                 // yarn 的 application id
  final val F_START_TIME    : String = "f_start_time"             // yarn 的 application 提交时间
  final val F_END_TIME      : String = "f_end_time"               // yarn 的 application 结束时间（只能为正常结束时间）
  final val F_REF_TIME      : String = "f_ref_time"               // 任务刷新时间，实时任务有效
  final val F_PARAM         : String = "f_param"                  // 任务参数
  final val F_ERR_CODE      : String = "f_err_code"               // 任务状态码

  override def getTBColumns: Seq[String] = Seq(
    F_TASK_NAME, F_TABLE_NAME, F_APP_ID, F_START_TIME, F_END_TIME,
    F_REF_TIME, F_PARAM, F_ERR_CODE)

  override def buildTable: String = {
    s"""
       |CREATE TABLE IF NOT EXISTS $getDBName.$getTBName (
       |$F_TASK_NAME VARCHAR NOT NULL
       |, $F_TABLE_NAME VARCHAR NOT NULL
       |, $F_APP_ID VARCHAR NOT NULL
       |, $F_START_TIME VARCHAR
       |, $F_END_TIME VARCHAR
       |, $F_REF_TIME VARCHAR
       |, $F_PARAM VARCHAR
       |, $F_ERR_CODE INTEGER
       |constraint pk primary key
       |($F_TASK_NAME, $F_TABLE_NAME, $F_APP_ID)) salt_buckets = $saltBuckets;
       |""".stripMargin
  }

  override def emptyTable: Dataset[_] = {
    import spark.implicits._
    spark.createDataset(spark.sparkContext.emptyRDD[TaskStatTable])
  }

  def createSingleRowDataset(taskStatTable: TaskStatTable): DataFrame = {
    val data = Array(taskStatTable)
    spark.createDataFrame(data).toDF(getTBColumns:_*)
  }

  case class TaskStatTable(f_task_name: String,
                           f_table_name: String,
                           f_app_id: String,
                           f_start_time: String,
                           f_end_time: String,    // 实时任务不会有结束时间
                           f_ref_time: String,    // 非实时任务不会有刷新时间
                           f_param: String,
                           f_error_code: Int)


  object TaskStatTable {

    def apply(row: Row): TaskStatTable = {
      TaskStatTable(row.getAs[String](F_TASK_NAME)
        , row.getAs[String](F_TABLE_NAME)
        , row.getAs[String](F_APP_ID)
        , row.getAs[String](F_START_TIME)
        , row.getAs[String](F_END_TIME)
        , row.getAs[String](F_REF_TIME)
        , row.getAs[String](F_PARAM)
        , row.getAs[Int](F_ERR_CODE)
      )
    }
  }
}
